<?php

namespace app\kernel\queue\amqp;

use app\kernel\Process;
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use think\Exception;
use think\facade\Config;
use think\helper\Arr;

class AmqpConsumerProcess extends Process
{

    /**
     * @var Process $process
     */
    protected $process;

    /**
     * @var AMQPStreamConnection
     */
    protected $amqp;

    /**
     * @var AbstractConsumer
     */
    protected $consumer;
    /**
     * @var string
     */
    private $prefix;

    /**
     * @return  void;
     * @throws Exception
     * @throws \ReflectionException
     */
    protected function beforeWorkerStart(Process $process): void
    {

        $this->process = $process;
    }

    /**
     * @return void
     */
    public function load(AbstractConsumer $abstractConsumer)
    {

        $this->consumer = $abstractConsumer;
    }

    /**
     * @return void
     * @throws \ErrorException
     */
    protected function workerStart(Process $process)
    {

        try {
            $this->registerLogger();
            //获取amqp 的链接
            $driver = $this->getChannelConfig($this->consumer->getConnect());
            if (empty($driver)) {
                $name = $this->getDefaultConnection();
                $driver = $this->getChannelConfig($name);
            }
            $prefix = $driver['prefix'];
            $connection = new AMQPStreamConnection($driver['host'], $driver['port'], $driver['user'], $driver['password'], $driver['vhost']);
            if (!$connection->isConnected()) {
                $connection->reconnect();
            }
            $exchange = $prefix . $this->consumer->getExchange(); //交换机
            $queue = $prefix . $this->consumer->getQueue();  //队列
            $routingKey = $prefix . $this->consumer->getQueue();  // 路由

            //获取频道
            $channel = $connection->channel();
            $channel->exchange_declare($exchange, $this->consumer->getType(), false, true);
            $channel->queue_declare($queue, false, true, false, false);
            $channel->queue_bind($queue, $exchange, $routingKey);
            $channel->basic_consume($queue, false, false, false, false, false, [
                $this,
                'consumer'
            ]);
            #检查结果
            while (count($channel->callbacks)) {
                try {
                    $channel->wait();
                } catch (\Throwable $exception) {
                    $this->out->error($exception->getMessage());
                    $this->app->log->error((string)$exception);
                    break;
                }
            }

        } catch (\Throwable $exception) {
            isset($channel) && $channel->close();
            isset($connection) && $connection->close();
            $this->out->error($exception->getMessage());
            $this->app->log->error((string)$exception);
        }
    }

    /**
     * @param AMQPMessage $msg
     *
     * @return void
     */
    public function consumer(AMQPMessage $msg)
    {

        /**
         * @var AMQPChannel $channel
         */
        $channel = $msg->delivery_info['channel'];
        $deliveryTag = $msg->delivery_info['delivery_tag'];

        #如果消息非json
        $message = json_decode($msg->body, true);
        if (json_last_error() == true || !is_array($message)) {
            $result = Result::DROP;
        } else {

            try {
                $result = $this->consumer->consumer($message);
                if (empty($result)) {
                    $result = Result::ACK;
                }
            } catch (\Throwable  $exception) {
                $this->out->error($exception->getMessage());
                $this->app->log->error($exception->getMessage());
                $result = Result::DROP;
            }

        }

        #使用完成后关闭mysql 链接
        $this->app->make('think\DbManager')->connect()->close();

        if ($result === Result::ACK) {
            return $channel->basic_ack($deliveryTag);
        }
        if ($result === Result::DROP) {
            return $channel->basic_reject($deliveryTag, false);
        }
        if ($result === Result::NACK) {
            return $channel->basic_nack($deliveryTag, false, $this->consumer->isRequeue());
        }
        if ($this->consumer->isRequeue() && $result === Result::REQUEUE) {
            return $channel->basic_reject($deliveryTag, true);
        }

        return $channel->basic_reject($deliveryTag, $this->consumer->isRequeue());

    }

    /**
     * 获取默认驱动
     *
     * @return mixed
     */
    public function getDefaultConnection()
    {

        return $this->getConfig('default');
    }

    /**
     * 获取日志配置
     *
     * @access public
     *
     * @param null|string $name 名称
     * @param mixed $default 默认值
     *
     * @return mixed
     */
    public function getConfig(string $name = NULL, $default = NULL)
    {

        if (!is_null($name)) {
            return $this->app->config->get('amqp.' . $name, $default);
        }

        return $this->app->config->get('amqp');
    }

    /**
     * 获取渠道配置
     *
     * @param string $channel
     * @param null $name
     * @param null $default
     *
     * @return array
     */
    public function getChannelConfig($channel, $name = NULL, $default = NULL): array
    {

        if ($config = $this->getConfig("connections.{$channel}")) {
            return Arr::get($config, $name, $default);
        }

        return [];
    }

    /**
     * 注册日志
     *
     * @param $class
     *
     * @return void
     */
    protected function registerLogger()
    {

        $config = Config::get('log.channels.file');
        $config['path'] = app()->getRuntimePath() . 'log/queueAmqp';
        $channels = array_merge(Config::get('log.channels'), ['queue_amqp' => $config]);
        Config::set(['channels' => $channels], 'log');
        Config::set(['default' => 'queue_amqp'], 'log');
    }
}